package gossip;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

public class FailureDetector {

    private final GossipCore gossipCore;
    private final Map<String, Long> lastHeardTimes = new ConcurrentHashMap<>();
    private static final long SUSPECT_THRESHOLD = 3000; // 3秒无响应视为可疑

    public FailureDetector(GossipCore gossipCore) {
        this.gossipCore = gossipCore;
    }

    public void check() {
        long now = System.currentTimeMillis();
        Map<String, MemberState> members = gossipCore.getMembershipTable().getMembers();

        // 更新本地节点时间戳
        gossipCore.getMembershipTable().getLocalState().updateTimestamp();

        // 检测可疑节点
        for (MemberState member : members.values()) {
            if (member == gossipCore.getMembershipTable().getLocalState()) continue;

            Long lastHeard = lastHeardTimes.get(member.getNodeId());
            if (lastHeard == null) {
                lastHeardTimes.put(member.getNodeId(), now);
            } else if (now - lastHeard > SUSPECT_THRESHOLD) {
                handleSuspectNode(member.getNodeId());
            }
        }
    }

    private void handleSuspectNode(String nodeId) {
        System.out.println("Node " + nodeId + " is suspected down");

        // 间接检查：询问其他节点
        List<MemberState> targets = gossipCore.getMembershipTable()
                .selectRandomTargets(2)
                .stream()
                .filter(m -> !m.getNodeId().equals(nodeId))
                .collect(Collectors.toList());

        for (MemberState target : targets) {
            String probeMsg = "PROBE:" + nodeId;
            gossipCore.getTransport().send(target.getAddress(), probeMsg);
        }
    }

    public void recordHeartbeat(String nodeId) {
        lastHeardTimes.put(nodeId, System.currentTimeMillis());
    }
}
